package com.kafka.kafkatest.service;

import cn.hutool.json.JSONObject;
import com.kafka.kafkatest.server.WebSocketServer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;

@Service
@Slf4j
public class DataProcessingService {


    @Autowired
    private WebSocketServer webSocketServer;

    @Autowired
    private RedisTemplate redisTemplate;

    @Async("taskExecutor")
    public void processData(String data) {
        //发送消息正在处理中
        JSONObject msg=new JSONObject();
        msg.set("code","1");
        msg.set("msg","正在处理中");
        webSocketServer.sendInfo( msg.toString(),"11");
        System.out.println("=================正在执行=================="+data);
        // 这里是处理数据的代码，它将在线程池中运行
        // 假设处理数据的逻辑
        try {
            Thread.sleep(8000); // 模拟数据处理需要一定时间
             log.info("Processing data: " + data);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        ValueOperations operations = redisTemplate.opsForValue();
        //正在处理的数据总数 -1
        operations.decrement("num",1);
        msg.set("code","2");
        msg.set("msg","处理完成");
        msg.set("data","================");
        webSocketServer.sendInfo(msg.toString(),"11");
    }
}
